package org.groupg.practice;


import io.reactivex.rxjava3.core.BackpressureOverflowStrategy;
import io.reactivex.rxjava3.core.Flowable;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.schedulers.Schedulers;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class RxJavaExample {

    public static void main(String[] args) throws InterruptedException {
        // 1. 创建型操作符
        createObservable();
        Thread.sleep(1000);

        // 2. 转换操作符
        transformOperators();
        Thread.sleep(1000);

        // 3. 过滤操作符
        filterOperators();
        Thread.sleep(1000);

        // 4. 组合操作符
        combiningOperators();
        Thread.sleep(1000);

        // 5. 错误处理
        errorHandling();
        Thread.sleep(1000);

        // 6. 线程调度
        threadingExample();
        Thread.sleep(1000);

        // 7. 背压处理
        backpressureHandling();
        Thread.sleep(5000); // 确保所有异步任务完成
    }

    // 1. 创建型操作符示例
    private static void createObservable() {
        System.out.println("\n===== 创建型操作符 =====");

        // just: 直接发射预定义的数据项
        Observable.just("Apple", "Banana", "Cherry")
                .subscribe(item -> System.out.println("just: " + item));

        // fromIterable: 从集合发射数据
        List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5);
        Observable.fromIterable(numbers)
                .subscribe(num -> System.out.println("fromIterable: " + num));

        // interval: 定时发射递增数字
        Disposable intervalDisposable = Observable.interval(500, TimeUnit.MILLISECONDS)
                .take(3) // 只取3个
                .subscribe(
                        tick -> System.out.println("interval: " + tick),
                        Throwable::printStackTrace,
                        () -> System.out.println("interval 完成!")
                );
    }

    // 2. 转换操作符示例
    private static void transformOperators() {
        System.out.println("\n===== 转换操作符 =====");

        // map: 类型转换
        Observable.just("Hello", "World")
                .map(String::toUpperCase)
                .subscribe(str -> System.out.println("map: " + str));

        // flatMap: 扁平化转换
        Observable.just("A,B,C", "D,E,F")
                .flatMap(s -> Observable.fromArray(s.split(",")))
                .subscribe(letter -> System.out.println("flatMap: " + letter));

        // concatMap: 保持顺序的扁平化
        Observable.just(100, 200, 300)
                .concatMap(delay -> Observable.just(delay)
                        .delay(delay, TimeUnit.MILLISECONDS))
                .subscribe(item -> System.out.println("concatMap: " + item));

        // buffer: 缓冲数据
        Observable.range(1, 10)
                .buffer(3)
                .subscribe(list -> System.out.println("buffer: " + list));
    }

    // 3. 过滤操作符示例
    private static void filterOperators() {
        System.out.println("\n===== 过滤操作符 =====");

        // filter: 条件过滤
        Observable.just(1, 2, 3, 4, 5, 6)
                .filter(num -> num % 2 == 0)
                .subscribe(even -> System.out.println("filter: " + even));

        // take: 取前N项
        Observable.interval(200, TimeUnit.MILLISECONDS)
                .take(3)
                .subscribe(item -> System.out.println("take: " + item));

        // skip: 跳过前N项
        Observable.range(1, 5)
                .skip(2)
                .subscribe(num -> System.out.println("skip: " + num));

        // distinct: 去重
        Observable.just(1, 2, 2, 3, 3, 3)
                .distinct()
                .subscribe(unique -> System.out.println("distinct: " + unique));
    }

    // 4. 组合操作符示例
    private static void combiningOperators() {
        System.out.println("\n===== 组合操作符 =====");

        // zip: 多流合并
        Observable<Integer> numbers = Observable.just(1, 2, 3);
        Observable<String> letters = Observable.just("A", "B", "C");

        Observable.zip(numbers, letters,
                (num, letter) -> num + letter)
                .subscribe(comb -> System.out.println("zip: " + comb));

        // merge: 合并多个流（可能交错）
        Observable<Long> fast = Observable.interval(100, TimeUnit.MILLISECONDS).take(3);
        Observable<Long> slow = Observable.interval(300, TimeUnit.MILLISECONDS).take(2);

        Observable.merge(fast, slow)
                .subscribe(item -> System.out.println("merge: " + item));

        // concat: 顺序连接流
        Observable.concat(
                Observable.just("First"),
                Observable.just("Second"),
                Observable.just("Third")
        ).subscribe(item -> System.out.println("concat: " + item));
    }

    // 5. 错误处理示例
    private static void errorHandling() {
        System.out.println("\n===== 错误处理 =====");

        // onErrorReturn: 遇到错误返回默认值
        Observable.error(new RuntimeException("模拟错误"))
                .onErrorReturn(throwable -> {
                    System.out.println("捕获错误: " + throwable.getMessage());
                    return -1;
                })
                .subscribe(
                        item -> System.out.println("onErrorReturn: " + item),
                        err -> System.err.println("这行不应该执行")
                );

        // retry: 重试机制
        AtomicInteger counter = new AtomicInteger();
        Observable.create(emitter -> {
                    if (counter.incrementAndGet() < 3) {
                        System.out.println("暂时错误");
                        emitter.onError(new RuntimeException("暂时错误"));
                    } else {
                        System.out.println("成功数据");
                        emitter.onNext("成功数据");
                        emitter.onComplete();
                    }
                })
                .retry(2) // 重试2次
                .subscribe(
                        data -> System.out.println("retry成功: " + data),
                        err -> System.out.println("最终失败: " + err.getMessage())
                );
    }

    // 6. 线程调度示例
    private static void threadingExample() {
        System.out.println("\n===== 线程调度 =====");

        Observable.create(emitter -> {
                    System.out.println("发射线程: " + Thread.currentThread().getName());
                    emitter.onNext(1);
                    emitter.onComplete();
                })
                .subscribeOn(Schedulers.io()) // 指定发射线程
                .observeOn(Schedulers.computation()) // 指定观察线程
                .doOnNext(i -> System.out.println("处理线程: " + Thread.currentThread().getName()))
                .subscribe();
    }

    // 7. 背压处理示例
    private static void backpressureHandling() {
        System.out.println("\n===== 背压处理 =====");

        Flowable.interval(10, TimeUnit.MILLISECONDS)
                .onBackpressureBuffer(10, // 缓冲区大小
                        () -> System.out.println("缓冲区溢出!"),
                        BackpressureOverflowStrategy.DROP_LATEST)
                .observeOn(Schedulers.single(), false, 5) // 设置接收缓冲区大小
                .subscribe(
                        item -> {
                            Thread.sleep(100); // 模拟慢消费者
                            System.out.println("处理: " + item);
                        },
                        Throwable::printStackTrace
                );
    }
}